package cn._51doit.live.udfs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.utils.EventType;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.Map;

/**
 * 要求观众停留一分钟，才算是一个有效的观众
 */
public class AudienceCountFunctionV2 extends KeyedProcessFunction<String, DataBean, Tuple3<String, Integer, Integer>> {

    //mapState中用来装DeviceId和用户进入直播间对应的时间
    private transient MapState<String, Long> mapState;

    private transient ValueState<Integer> totalCountState;
    private transient ValueState<Integer> onlineCountState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Integer> stateDescriptor1 = new ValueStateDescriptor<>("total-count-state", Integer.class);
        totalCountState = getRuntimeContext().getState(stateDescriptor1);

        ValueStateDescriptor<Integer> stateDescriptor2 = new ValueStateDescriptor<>("online-count-state", Integer.class);
        onlineCountState = getRuntimeContext().getState(stateDescriptor1);

        MapStateDescriptor<String, Long> stateDescriptor = new MapStateDescriptor<>("uid-time-state", String.class, Long.class);
        mapState = getRuntimeContext().getMapState(stateDescriptor);
    }

    @Override
    public void processElement(DataBean value, Context ctx, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
        String deviceId = value.getDeviceId();

        String eventId = value.getEventId();

        Integer onlineCount = onlineCountState.value();
        if (onlineCount == null) {
            onlineCount = 0;
        }


        //进入
        if(EventType.LIVE_ENTER.equals(eventId)) {
            //统计停留一分钟的才是有效用户，计入的事件，就注册定时器（Timer）
            long currentTime = System.currentTimeMillis();
            ctx.timerService().registerProcessingTimeTimer(currentTime + 60000);
            //将用户的DeviceId和进入的实现保存到MapState中
            mapState.put(deviceId, currentTime);

            //2.统计在线用户（进入）
            onlineCount +=1;

        } else {
            //统计在线用户（离开）
            onlineCount -= 1;

            Long enterTime = mapState.get(deviceId);
            if (enterTime != null) {
                //停留没有超过1分钟，定时器还没执行，用户就厉离开了
                if(System.currentTimeMillis() - enterTime < 60000) {
                    mapState.remove(deviceId);
                }
            }
        }
        //更新
        onlineCountState.update(onlineCount);

        //mapState.put("ONLINE_COUNT", onlineCount);
        //输出数据


        out.collect(Tuple3.of(ctx.getCurrentKey(), totalCountState.value() == null ? 0 : totalCountState.value() , onlineCount));

    }


    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
        Integer totalCount = totalCountState.value();
        if (totalCount == null) {
            totalCount = 0;
        }
        //判断是否大于一分钟
        Iterator<Map.Entry<String, Long>> iterator = mapState.iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Long> entry = iterator.next();
            //String deviceId = entry.getKey();
            Long enterTime = entry.getValue();
            long current = System.currentTimeMillis();
            if (current - enterTime >= 60000) {
                //才是一个有效的用户
                //有效用户数量+1
                //累加状态
                totalCount += 1;

                //将当前遍历的的数据移除（当前设备ID的，避免重复计算）
                iterator.remove();
            }
        }
        //更新状态
        totalCountState.update(totalCount);

    }
}
